package com.navinfo.platform.bigdata.hive.udtf;

import com.alibaba.fastjson.JSON;
import com.navinfo.platform.bigdata.hive.dto.FaultInfoDto;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hive.ql.exec.Description;
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

import java.util.ArrayList;
import java.util.List;

/**
 * 将0f39故障信息faultlist字段，转换成多行
 */
@Description(
        name = "faultlist",
        value = "_FUNC_(col) - The parameter is a column name. The return value is two strings.",
        extended = "Example:\n"
                + " > SELECT _FUNC_(col) FROM src;"
                + " > SELECT _FUNC_(col) AS (name, surname) FROM src;"
                + " > SELECT adTable.name,adTable.surname FROM src LATERAL VIEW _FUNC_(col) adTable AS name, surname;"
)
public class FaultUDF extends GenericUDTF {
    private static List<Integer[]> unpackFaultInfo(List<FaultInfoDto> dataList) {
        final List<Integer[]> integers = new ArrayList<>();
        for (FaultInfoDto faultInfoDto : dataList) {
            Integer[] result = new Integer[]{0, 0, 0, 0, 0,
                    0, 0, 0, 0, 0,
                    0, 0, 0, 0, 0,
                    0, 0, 0, 0, 0};
            result[0] = faultInfoDto.getFaultType();
            result[1] = faultInfoDto.getSubSpeed();
            result[2] = faultInfoDto.getAccelerator();
            result[3] = faultInfoDto.getBrakeSignal();
            result[4] = faultInfoDto.getRotation();

            result[5] = faultInfoDto.getEngineTurbochargedPressure();
            result[6] = faultInfoDto.getEngineIntakePressure();
            result[7] = faultInfoDto.getEngineExhaustTemperature();
            result[8] = faultInfoDto.getEngineCoolantTemperature();
            result[9] = faultInfoDto.getAccelerationRate();

            result[10] = faultInfoDto.getGear();
            result[11] = faultInfoDto.getEngineOutputTorque();
            result[12] = faultInfoDto.getLoad();
            result[13] = faultInfoDto.getEngineLoad();
            result[14] = faultInfoDto.getVehicleAcceleration();

            result[15] = faultInfoDto.getVehicleSpeedReduction();
            result[16] = faultInfoDto.getFaultCodeCount();

            final List<FaultInfoDto.FaultItem> faultItemList = faultInfoDto.getFaultItemList();
            if (faultItemList.size() > 0) {
                for (FaultInfoDto.FaultItem faultItem : faultItemList) {
                    Integer[] dest = new Integer[result.length];
                    System.arraycopy(result, 0, dest, 0, result.length);
                    dest[17] = faultItem.getFaultAddr();
                    dest[18] = faultItem.getSPN();
                    dest[19] = faultItem.getFMI();
                    integers.add(dest);
                }
            } else {
                integers.add(result);
            }
        }
        return integers;
    }

    /**
     * 返回类型为 String，string
     *
     * @param argOIs
     * @return
     * @throws UDFArgumentException
     */
    @Override
    public StructObjectInspector initialize(StructObjectInspector argOIs) throws UDFArgumentException {

        if (argOIs.getAllStructFieldRefs().size() != 1) {
            throw new UDFArgumentException("StatusAdditionUDF takes exactly one argument.");
        }
        final ObjectInspector fieldObjectInspector = argOIs.getAllStructFieldRefs().get(0).getFieldObjectInspector();
        if (fieldObjectInspector.getCategory() != ObjectInspector.Category.PRIMITIVE
                && ((PrimitiveObjectInspector) fieldObjectInspector).getPrimitiveCategory() != PrimitiveObjectInspector.PrimitiveCategory.STRING) {
            throw new UDFArgumentTypeException(0, "StatusAdditionUDF takes a string as a parameter.");
        }

        List<String> fieldNames = new ArrayList<>();
        fieldNames.add("faultType");
        fieldNames.add("subSpeed");
        fieldNames.add("accelerator");
        fieldNames.add("brakeSignal");
        fieldNames.add("rotation");

        fieldNames.add("engineTurbochargedPressure");
        fieldNames.add("engineIntakePressure");
        fieldNames.add("engineExhaustTemperature");
        fieldNames.add("engineCoolantTemperature");
        fieldNames.add("accelerationRate");

        fieldNames.add("gear");
        fieldNames.add("engineOutputTorque");
        fieldNames.add("load");
        fieldNames.add("engineLoad");
        fieldNames.add("vehicleAcceleration");

        fieldNames.add("vehicleSpeedReduction");
        fieldNames.add("faultCodeCount");
        fieldNames.add("faultAddr");
        fieldNames.add("SPN");
        fieldNames.add("FMI");

        List<ObjectInspector> fieldOIs = new ArrayList<>();
        for (int i = 0; i < fieldNames.size(); i++) {
            fieldOIs.add(PrimitiveObjectInspectorFactory.javaIntObjectInspector);
        }
        return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs);
    }

    /**
     * 该方法处理输入记录，然后通过forward()方法返回输出结果
     *
     * @param args
     * @throws HiveException
     */
    @Override
    public void process(Object[] args) throws HiveException {
        String input = args[0].toString();
        if (StringUtils.startsWith(input, "[")) {
            List<FaultInfoDto> dataList = JSON.parseArray(input, FaultInfoDto.class);
            List<Integer[]> results = unpackFaultInfo(dataList);
            for (Integer[] re : results) {
                forward(re);
            }
        }
    }

    /**
     * 用于通知UDTF没有行可以处理了。可以在该方法中清理代码或者附加其他处理输出。
     *
     * @throws HiveException
     */
    @Override
    public void close() throws HiveException {
    }
}
